<?php

declare(strict_types=1);

namespace Plugin\Alarm\Process;

use Common\Common;
use Hyperf\Coroutine\Coroutine;
use Hyperf\Process\AbstractProcess;
use Hyperf\Process\Annotation\Process;
use Plugin\Alarm\Repository\RdAlarmWorkRepository;
use Plugin\LModel\Service\LModelConfigsService;

#[Process(name: 'alarm_review_consumer')]
class ReviewConsumerProcess extends AbstractProcess
{
    const STREAM_NAME = 'alarm:review';
    const GROUP_NAME = 'work_reviewer';

    public function handle(): void
    {
        try {
//            检查是否可获取大模型
            Common::container()->get(LModelConfigsService::class);
        } catch (\Throwable $exception) {
            return;
        }
        // 确保消费者组存在
        Common::redisStreamCreateGroup(self::STREAM_NAME, self::GROUP_NAME);
        
        while (true) {
            try {
                // 阻塞模式监听消息
                $messages = Common::redisStreamConsume(
                    self::STREAM_NAME,
                    self::GROUP_NAME,
                    'consumer1', 
                    3,
                    3000
                );
                
                if (!empty($messages)) {
                    foreach ($messages as $stream => $streamMessages) {
                        foreach ($streamMessages as $messageId => $message) {
                            // 处理消息
                            $this->processMessage($messageId, $message);
                            Coroutine::sleep(3);

                        }
                    }
                }
            } catch (\Throwable $exception) {
                // 记录异常
                Common::logException(__CLASS__, $exception);
            }
        }
    }
    
    private function processMessage(string $messageId, array $message): void
    {
        try {
            $work = Common::container()->get(RdAlarmWorkRepository::class)->findById($message['work_id']);
            $extra = empty($work->extra) ? ['message_id' => $messageId] : array_merge($work->extra, ['message_id' => $messageId]);
            $work->update(['extra' => $extra]);
            Common::redisStreamAck(ReviewConsumerProcess::STREAM_NAME, ReviewConsumerProcess::GROUP_NAME, [$work['extra']['message_id']]);
//            @todo 调用大模型处理
        } catch (\Throwable $exception) {
            Common::logException(__CLASS__, $exception);
            return;
        }
    }

    public function isEnable($server): bool
    {
        return env('ENABLE_PROCESS', false);
    }
}
